# +----------------------------------------------------------------------
# | ChatWork智能聊天办公系统
# +----------------------------------------------------------------------
# | 软件声明: 本系统并非自由软件,未经授权任何形式的商业使用均属非法。
# | 版权保护: 任何企业和个人不允许对程序代码以任何形式任何目的复制/分发。
# | 授权要求: 如有商业使用需求,请务必先与版权所有者取得联系并获得正式授权。
# +----------------------------------------------------------------------
# | Author: ChatWork Team <2474369941@qq.com>
# +----------------------------------------------------------------------
import time
from typing import List
from tortoise.queryset import Q
from pydantic import TypeAdapter
from hypertext import PagingResult
from exception import AppException
from common.enums.pool import PoolEnum
from common.enums.voice import VoiceEnum
from common.enums.apply import VideoEnum
from common.enums.apply import MusicEnum
from common.enums.apply import DrawsEnum
from common.utils.urls import UrlUtil
from common.cache.pool_cache import KeyPoolCache
from common.models.dev import DevModelsModel
from common.models.dev import DevModelsKeysModel
from apps.admin.schemas.setting.ai import secret_schema as schema


class AiSecretService:

    @classmethod
    async def channel(cls, scene: str) -> List[schema.AiPoolKeysChannelVo]:
        """
        密钥通道

        Args:
            scene (str): 场景: [chat,vector,ranking,draws,...]

        Returns:
            List[schema.AiPoolKeysChannelVo]

        Author:
            zero
        """
        data = []
        if scene in [PoolEnum.TYPE_CHAT, PoolEnum.TYPE_VECTOR, PoolEnum.TYPE_RANKING]:
            orders = ["sort", "is_disable", "-id"]
            lists = await (DevModelsModel.filter(scene=scene, is_delete=0).order_by(*orders).all())
            for item in lists:
                data.append(schema.AiPoolKeysChannelVo(
                    name=item.name,
                    mark=item.mark,
                    channel=item.channel,
                    logo=await UrlUtil.to_absolute_url(item.logo),
                    is_disable=bool(item.is_disable)
                ))
        elif scene in [PoolEnum.TYPE_VOICE_SPEECH, PoolEnum.TYPE_VOICE_TRANSFER]:
            for k, v in VoiceEnum.get_channels().items():
                if k != "browser":
                    data.append(schema.AiPoolKeysChannelVo(
                        name=v,
                        mark=k,
                        channel=k,
                        logo=await UrlUtil.to_absolute_url(VoiceEnum.get_logo(k))
                    ))
        elif scene == PoolEnum.TYPE_DRAWS:
            for k, v in DrawsEnum.get_channels().items():
                data.append(schema.AiPoolKeysChannelVo(
                    name=v,
                    mark=k,
                    channel=k,
                    logo=await UrlUtil.to_absolute_url(DrawsEnum.get_logo(k))
                ))
        elif scene == PoolEnum.TYPE_VIDEO:
            for k, v in VideoEnum.get_channels().items():
                data.append(schema.AiPoolKeysChannelVo(
                    name=v,
                    mark=k,
                    channel=k,
                    logo=await UrlUtil.to_absolute_url(VideoEnum.get_logo(k))
                ))
        elif scene == PoolEnum.TYPE_MUSIC:
            for k, v in MusicEnum.get_channels().items():
                data.append(schema.AiPoolKeysChannelVo(
                    name=v,
                    mark=k,
                    channel=k,
                    logo=await UrlUtil.to_absolute_url(MusicEnum.get_logo(k))
                ))

        return data

    @classmethod
    async def lists(cls, params: schema.AiPoolKeysSearchIn) -> PagingResult[schema.AiPoolKeysListVo]:
        """
        密钥列表

        Args:
            params (schema.AiPoolKeysSearchIn): 搜索参数

        Returns:
            PagingResult[schema.AiPoolKeysListVo]

        Author:
            zero
        """
        where = [
            Q(channel=params.channel),
            Q(scene=params.scene),
            Q(is_delete=0)
        ]
        if params.keyword:
            where.append(Q(api_key__icontains=params.keyword))
        if params.status is not None:
            where.append(Q(is_disable=params.status))

        _model = DevModelsKeysModel.filter(*where).order_by("-sort", "-id")
        _pager = await DevModelsKeysModel.paginate(
            model=_model,
            page_no=params.page_no,
            page_size=params.page_size,
            fields=DevModelsKeysModel.without_field("is_delete,delete_time")
        )

        _lists = [TypeAdapter(schema.AiPoolKeysListVo).validate_python(item) for item in _pager.lists]
        _pager.lists = _lists
        return _pager

    @classmethod
    async def detail(cls, id_: int) -> schema.AiPoolKeysDetailVo:
        """
        密钥详情

        Args:
            id_ (int): ID

        Returns:
            schema.AiPoolKeysDetailVo

        Author:
            zero
        """
        detail = await DevModelsKeysModel.filter(id=id_).get()
        return TypeAdapter(schema.AiPoolKeysDetailVo).validate_python(detail.__dict__)

    @classmethod
    async def add(cls, post: schema.AiPoolKeysAddIn) -> None:
        """
        密钥新增

        Args:
            post (schema.AiPoolKeysAddIn): 参数

        Author:
            zero
        """
        # 验证模型通道
        if post.scene in [PoolEnum.TYPE_CHAT, PoolEnum.TYPE_VECTOR, PoolEnum.TYPE_RANKING]:
            post.channel = post.mark
            model = await DevModelsModel.filter(scene=post.scene, mark=post.mark, is_delete=0).first()
            if model is None:
                raise AppException("模型通道不存在,请刷新页面!")

        # 创建密钥数据
        await DevModelsKeysModel.create(
            scene=post.scene,
            channel=post.channel,
            app_id=post.app_id,
            api_key=post.api_key,
            api_secret=post.api_secret,
            sort=post.sort,
            is_disable=post.is_disable,
            create_time=int(time.time()),
            update_time=int(time.time())
        )

        # 删除密钥缓存
        await KeyPoolCache(scene=post.scene, channel=post.channel).del_key()

    @classmethod
    async def edit(cls, post: schema.AiPoolKeysEditIn) -> None:
        """
        密钥编辑

        Args:
            post (schema.AiPoolKeysAddIn): 参数

        Author:
            zero
        """
        # 验证密钥数据
        key = await DevModelsKeysModel.filter(id=post.id, is_delete=0).first()
        if key is None:
            raise AppException("密钥已不存在,请刷新页面!")

        # 验证模型通道
        if post.scene in [PoolEnum.TYPE_CHAT, PoolEnum.TYPE_VECTOR, PoolEnum.TYPE_RANKING]:
            post.channel = post.mark
            model = await DevModelsModel.filter(scene=post.scene, mark=post.mark, is_delete=0).first()
            if model is None:
                raise AppException("模型通道不存在,请刷新页面!")

        # 更新密钥数据
        error = key.error if post.is_disable else ""
        await DevModelsKeysModel.filter(id=post.id).update(
            scene=post.scene,
            channel=post.channel,
            app_id=post.app_id,
            api_key=post.api_key,
            api_secret=post.api_secret,
            error=error,
            sort=post.sort,
            is_disable=post.is_disable,
            update_time=int(time.time())
        )

        # 删除密钥缓存
        await KeyPoolCache(scene=post.scene, channel=post.channel).del_key()

    @classmethod
    async def delete(cls, id_: int) -> None:
        """
        密钥删除

        Args:
            id_ (int): ID

        Author:
            zero
        """
        # 验证密钥数据
        key = await DevModelsKeysModel.filter(id=id_, is_delete=0).first()
        if key is None:
            raise AppException("密钥已不存在,请刷新页面!")

        # 删除密钥数据
        await DevModelsKeysModel.filter(id=id_).update(
            is_delete=1,
            delete_time=int(time.time())
        )

        # 删除密钥缓存
        await KeyPoolCache(scene=key.scene, channel=key.channel).del_key()

    @classmethod
    async def disable(cls, id_: int) -> None:
        """
        密钥停用

        Args:
            id_ (int): ID

        Author:
            zero
        """
        # 验证密钥数据
        key = await DevModelsKeysModel.filter(id=id_, is_delete=0).first()
        if key is None:
            raise AppException("密钥已不存在,请刷新页面!")

        # 更新密钥数据
        await DevModelsKeysModel.filter(id=id_).update(
            is_disable=0 if key.is_disable else 1,
            delete_time=int(time.time())
        )

        # 删除密钥缓存
        await KeyPoolCache(scene=key.scene, channel=key.channel).del_key()
